package com.heima.schedule.service.Impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.constants.WmNewsConstants;
import com.heima.redis.CacheService;
import com.heima.schedule.dtos.Task;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.pojos.Taskinfo;
import com.heima.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.service.ITaskService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class TaskServiceImpl implements ITaskService {
    @Autowired
    private CacheService cacheService;

    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    private static final String FUTURE = "FUTURE";

    @Override
    public Long addTask(Task task) {
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        taskinfoMapper.insert(taskinfo);
        task.setTaskId(taskinfo.getTaskId());


        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo, taskinfoLogs);
        taskinfoLogs.setStatus(TaskinfoLogs.SCHEDULED);
        taskinfoLogs.setVersion(1);
        taskinfoLogsMapper.insert(taskinfoLogs);

        saveRedisTask(task);
        return null;
    }

    /**
     * 保存任务到redis
     *
     * @param task
     */
    private void saveRedisTask(Task task) {
        //获取当前时间
        Date now = new Date();
        //获取任务执行时间
        Date executeTime = task.getExecuteTime();
        //获取五分钟的时间
        Date minutes = DateUtils.addMinutes(now, 5);
        //构建key
        String key = buildTaskKey(task.getTaskType());


        if (executeTime.getTime() <= now.getTime()) {
            log.info("任务执行时间小于等于当前时间，发送MQ执行");
            //如果任务的执行时间小于等于当前时间，发送MQ执行
            sendExecMQ(task, key);
        } else if (executeTime.getTime() <= minutes.getTime()) {
            log.info("任务执行时间小于等于预设时间，存入zset中");
            //如果任务的执行时间小于或等于预设时间（未来5分钟） 存入zset中
            cacheService.zAdd(key, task.getTaskId() + "", task.getExecuteTime().getTime());
        }
    }

    /**
     * 定时任务
     */
    @Scheduled(cron = "0 */1 * * * ?")
    public void schedule() {
        log.info("定时任务执行");
        Boolean flag = redisTemplate.opsForValue().setIfAbsent("schedule", "1", 15, TimeUnit.SECONDS);
        try {
            if (!flag) {
                log.info("没有获取到锁,不执行");
                return;
            }
            Set<String> keys = cacheService.scan(FUTURE + "*");
            if (CollectionUtils.isEmpty(keys)) {
                log.info("没有延迟队列");
                return;
            }
            for (String key : keys) {
                //获取延迟任务
                Set<String> taskIds = cacheService.zRangeByScore(key, 0, System.currentTimeMillis());
                if (CollectionUtils.isEmpty(taskIds)) {
                    log.info("当前KEY:{}没有需要执行的任务", key);
                    continue;
                }
                log.info("当前KEY:{}需要执行的任务:{}", key, taskIds);
                for (String taskId : taskIds) {
                    log.info("执行任务:{}", taskId);
                    //根据taskId查询
                    Taskinfo taskinfo = taskinfoMapper.selectById(taskId);
                    Task task = new Task();
                    BeanUtils.copyProperties(taskinfo, task);
                    task.setTaskId(Long.valueOf(taskId));
                    //发送MQ消息
                    sendExecMQ(task, key);
                }
            }
        } finally {
            redisTemplate.delete("schedule");
        }

    }

    /**
     * 发送MQ消息
     *
     * @param task
     */
    private void sendExecMQ(Task task, String key) {
        log.info("发送MQ消息:{}", task);
        kafkaTemplate.send(WmNewsConstants.LEIMA_TASK_EXECUTOR, task.getParameters());
        //将task表数据删除，log表状态更新
        taskinfoMapper.deleteById(task.getTaskId());

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        taskinfoLogs.setTaskId(task.getTaskId());
        taskinfoLogs.setStatus(TaskinfoLogs.EXECUTED);
        taskinfoLogsMapper.updateById(taskinfoLogs);

        //zset数据移除
        cacheService.zRemove(key, task.getTaskId() + "");
    }

    /**
     * 数据库任务同步到Redis
     */
    @Scheduled(cron = "0 */1 * * * ?")
    public void syncTask() {
        log.info("同步任务执行");
        Boolean flag = redisTemplate.opsForValue().setIfAbsent("syncTask", "1", 5, TimeUnit.MINUTES);
        try {
            if (!flag) {
                log.info("没有获取到锁,不执行");
                return;
            }
            Date now = new Date();
            //获取未来5分钟
            Date minutes = DateUtils.addMinutes(now, 5);
            //查询当前时间并且小于未来5分钟的任务
            LambdaQueryWrapper<Taskinfo> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.le(Taskinfo::getExecuteTime, minutes);
            //查询
            List<Taskinfo> taskinfoList = taskinfoMapper.selectList(queryWrapper);
            //判断是否为空
            if (CollectionUtils.isEmpty(taskinfoList)) {
                log.info("没有需要同步的任务");
                return;
            }
            log.info("需要同步的任务:{}", taskinfoList);
            //遍历,同步到Redis中
            for (Taskinfo taskinfo : taskinfoList) {
                Task task = new Task();
                BeanUtils.copyProperties(taskinfo, task);
                task.setTaskId(taskinfo.getTaskId());
                saveRedisTask(task);
            }
            log.info("同步任务执行完成");
        } finally {
            redisTemplate.delete("syncTask");
        }


    }

    /**
     * key
     *
     * @param taskType
     * @return
     */
    private String buildTaskKey(Integer taskType) {
        return "FUTURE" + taskType;
    }
}
